#pragma once
#include <google/protobuf/service.h>
#include "fastrpc/base.hpp"
#include "fastrpc/log.hpp"
#include "fastrpc/net.hpp"
#include "fastrpc/rpc/fastrpc/fastrpc_callback.hpp"
#include "fastrpc/rpc/fastrpc/fastrpc_controller.hpp"
#include "fastrpc/rpc/fastrpc/fastrpc_dispatcher.hpp"
#include "fastrpc/rpc/rpc_framed.hpp"

namespace fastrpc::rpc {
using namespace fastrpc::async;
using namespace fastrpc::net;
using namespace fastrpc::log;
using namespace fastrpc;
using namespace fastrpc::rpc;

class RpcServer;
auto process(TcpStream stream, RpcServer* server) -> Task<void>;

class RpcServer {
public:
    RpcServer(uint16_t port) : port_{port} {}

public:
    void register_service(std::shared_ptr<google::protobuf::Service> service) {
        rpc::detail ::RpcDispatcher::GetRpcDispatcher()->registerService(service);
    }

    void run() {
        runtime::MultiThreadBuilder::options().build().block_on([this]() -> Task<void> {
            auto has_addr = SocketAddr::parse("0.0.0.0", port_);
            if (!has_addr) {
                console.error(has_addr.error().message());
                co_return;
            }
            auto has_listener = TcpListener::bind(has_addr.value());
            if (!has_listener) {
                console.error(has_listener.error().message());
                co_return;
            }
            // console.info("Listening on {}:{} ...", "0.0.0.0", port_);
            auto listener = std::move(has_listener.value());
            while (true) {
                auto has_stream = co_await listener.accept();

                if (has_stream) {
                    auto& [stream, peer_addr] = has_stream.value();
                    // console.info("Accept a connection from {}", peer_addr);
                    spawn(this->handle_rpc(std::move(stream)));
                } else {
                    console.error(has_stream.error().message());
                    break;
                }
            }
        }());
    }
    void set_callback(FastRpcClosure* cb) {
        callback_ = cb;
    }

private:
    auto handle_rpc(TcpStream stream) -> Task<void> {
        std::shared_ptr<rpc::RpcFramed<>> rpc_framed = std::make_shared<rpc::RpcFramed<>>(stream);

        while (true) {
            std::array<char, 1024> buf{};
            auto read_result = co_await rpc_framed->read_frame(buf);  // 读取并解码
            if (!read_result) {
                console.info("client closed");
                co_return;
            }

            auto                  decoded_message = read_result.value();
            FastPBProtocol::s_ptr request = std::make_shared<FastPBProtocol>(decoded_message);
            FastPBProtocol::s_ptr reponse = std::make_shared<FastPBProtocol>();
            rpc::detail::RpcDispatcher::GetRpcDispatcher()->dispatch(request,
                                                                     reponse,
                                                                     stream,
                                                                     callback_);
            auto result = *reponse;
            auto write_result = co_await rpc_framed->write_frame(result);  // 编码并发送回复
            if (!write_result) {
                console.error("{}", write_result.error().message());
                co_return;
            }
        }
    }

private:
    uint16_t        port_;
    FastRpcClosure* callback_;
};
}  // namespace fastrpc::rpc
